package transport;

import hdfs.TransPort;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import transport.handler.InnerHandler;
import transport.handler.OuterHandler;

import java.util.concurrent.ArrayBlockingQueue;


public class WriteBlockTransPorter {

    private static final Logger LOG = LoggerFactory.getLogger(WriteBlockTransPorter.class);
    private final int PORT;
    private final String storagePath;
    private final ArrayBlockingQueue<ChannelContext> bridge;
    EventLoopGroup acceptEventLoopGroup;
    EventLoopGroup workEventLoopGroup;
    Channel remoteChannel;

    public WriteBlockTransPorter(int PORT,String storagePath){
        this.PORT = PORT;
        this.storagePath = storagePath;
        bridge  = new ArrayBlockingQueue<>(1);
    }

    public void startServer(){
        acceptEventLoopGroup = new NioEventLoopGroup();
        workEventLoopGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(acceptEventLoopGroup, workEventLoopGroup);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
                pipeline.addLast(new ProtobufVarint32FrameDecoder());
                pipeline.addLast(new ProtobufDecoder(TransPort.WritePacketRequest.getDefaultInstance()));
                pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
                pipeline.addLast(new ProtobufEncoder());
                pipeline.addLast(new OuterHandler(bridge,storagePath,acceptEventLoopGroup,workEventLoopGroup));
            }
        });
        bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
        try {
            bootstrap.bind(PORT).sync();
            LOG.info("success bind local address on {}", PORT);
        } catch (InterruptedException e) {
            LOG.error("failed bind local address on {}", PORT);
            e.printStackTrace();
        }
    }

    public synchronized void connect(String nextIp,int nextPort){
        ChannelContext relayCtx = new ChannelContext();
        try {
            bridge.put(relayCtx);
        } catch (InterruptedException e) {
            LOG.warn("connect fail to remote address{}", nextPort);
            e.printStackTrace();
        }
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workEventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
//                pipeline.addLast("decoder",new ProtobufDecoder(TransPort.WritePacketResponse.getDefaultInstance()));
                pipeline.addLast(new ProtobufVarint32FrameDecoder());
                pipeline.addLast(new ProtobufDecoder(TransPort.WritePacketRequest.getDefaultInstance()));
                pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
                pipeline.addLast(new ProtobufEncoder());
                pipeline.addLast(new InnerHandler(relayCtx));
//                System.out.println("建立子通道的地址"+System.identityHashCode(relayCtx));
            }
        });

        try {
            ChannelFuture ch = bootstrap.connect(nextIp, nextPort).sync();
            LOG.info("connect success to remote address {}", nextPort);
            relayCtx.setRemoteChannel(ch.channel());
        } catch (InterruptedException e) {
            LOG.warn("connect fail to remote address{}", nextPort);
            e.printStackTrace();
            shutDown();
        }
    }

    public void addBridge(){
        try {
            this.bridge.put(new ChannelContext());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void shutDown(){
        if(acceptEventLoopGroup!=null){
            acceptEventLoopGroup.shutdownGracefully();
        }
        if(workEventLoopGroup!=null){
            workEventLoopGroup.shutdownGracefully();
        }
        LOG.info("all stopped");
    }
}
